package cn.itcast.task;

import cn.itcast.bean.CleanBean;
import cn.itcast.bean.WarnBaseBean;
import cn.itcast.inter.ProcessDataInterface;
import cn.itcast.mail.MailSend;
import cn.itcast.util.RedisUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.cep.CEP;
import org.apache.flink.cep.PatternSelectFunction;
import org.apache.flink.cep.PatternStream;
import org.apache.flink.cep.pattern.Pattern;
import org.apache.flink.cep.pattern.conditions.SimpleCondition;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import redis.clients.jedis.JedisCluster;

import java.math.BigDecimal;
import java.util.List;
import java.util.Map;

/**
 * @Date 2021
 * 个股涨跌幅行情
 */
public class UpDownTask implements ProcessDataInterface {
    @Override
    public void process(DataStream<CleanBean> waterData) {
        /**
         * 1.数据转换map
         * 2.封装样例类数据
         * 3.加载redis涨跌幅数据
         * 4.模式匹配
         * 5.获取匹配模式流数据
         *   6.查询数据
         * 7.发送告警邮件
         */
        //1.数据转换map
        SingleOutputStreamOperator<WarnBaseBean> mapData = waterData.map(new MapFunction<CleanBean, WarnBaseBean>() {
            @Override
            public WarnBaseBean map(CleanBean value) throws Exception {
                WarnBaseBean warnBaseBean = new WarnBaseBean(
                        value.getSecCode(),
                        value.getPreClosePrice(),
                        value.getMaxPrice(),
                        value.getMinPrice(),
                        value.getTradePrice(),
                        value.getEventTime()
                );
                return warnBaseBean;
            }
        });
        //3.加载redis涨跌幅数据
        JedisCluster jedisCluster = RedisUtil.getJedisCluster();
        //跌幅限制
        BigDecimal upd1 = new BigDecimal(jedisCluster.hget("quot", "upDown1"));
        //涨幅限制
        BigDecimal upd2 = new BigDecimal(jedisCluster.hget("quot", "upDown2"));
        //模式匹配
        Pattern<WarnBaseBean, WarnBaseBean> upDown = Pattern.<WarnBaseBean>begin("upDown")
                .where(new SimpleCondition<WarnBaseBean>() {
                    //判断涨跌幅大小
                    //涨跌幅：(期末收盘点位-期初前收盘点位)/期初前收盘点位*100%
                    //计算实时流中得涨跌幅
                    @Override
                    public boolean filter(WarnBaseBean value) throws Exception {
                        BigDecimal divide = (value.getClosePrice().subtract(value.getPreClosePrice())).divide(value.getPreClosePrice(), 2, BigDecimal.ROUND_HALF_UP);
                        if (divide.compareTo(upd1) == 1 || divide.compareTo(upd2) == 1) {
                            return true;
                        }
                        return false;
                    }
                });
        //5.获取匹配模式流数据
        PatternStream<WarnBaseBean> cep = CEP.pattern(mapData.keyBy(WarnBaseBean::getSecCode), upDown);
        //6.查询数据
        cep.select(new PatternSelectFunction<WarnBaseBean, Object>() {
            @Override
            public Object select(Map<String, List<WarnBaseBean>> pattern) throws Exception {

                List<WarnBaseBean> list = pattern.get("upDown");
                if (list!=null&&list.size()>0){
                    //发送实时告警邮件
                    MailSend.send("涨跌幅实时告警:"+list.toString());
                }
                return list;
            }
        }).print();
    }
}
